package server

import (
	"fmt"
	"github.com/Shopify/sarama"
	"net"
	"os"
	log "socketTokafka/logger"
	kafka "socketTokafka/product"
	"strings"
)

/**
client ：echo “hello2222111111asdasdsa撒大大实打实的啊实打实大苏打阿萨大大撒大苏打飒飒的是覅点击返回酒店SDassssssssssssssssssssssssFJDSFKDASLFJKDSF DFSADSAF1” > /dev/udp/127.0.0.1/7777

./kafka-console-consumer.sh --bootstrap-server 10.65.66.181:9092 --topic test1_socket --from-beginning
*/
type Server struct {
	product   *kafka.Producer
	listenUDP string
	quit      chan os.Signal
	limitChan chan bool
}

func (s *Server) SetClient(c *kafka.Producer) {
	s.product = c
}

func (s *Server) GetClient() *kafka.Producer {
	return s.product
}
func NewServer(product *kafka.Producer, listenUDP string, goroutineNumber int) *Server {
	quit := make(chan os.Signal)
	return &Server{
		product:   product,
		listenUDP: listenUDP,
		quit:      quit,
		limitChan: make(chan bool, goroutineNumber),
	}
}

func (s *Server) SetListenUDP(listen string) {
	s.listenUDP = listen
}

func (s *Server) SetListenTCP(listen string) {
	s.listenUDP = listen
}

func (s *Server) udpProcess(conn *net.UDPConn, msga chan *sarama.ProducerMessage, msgb chan *sarama.ProducerMessage) {

	// 最大读取数据大小
	data := make([]byte, 16384)
	n, _, err := conn.ReadFromUDP(data)
	if err != nil {
		fmt.Println("productA failed read udp msg, error: " + err.Error())
	}
	str := string(data[:n])
	//newstr := str[:1] + "{" + str[1:]
	//index := len(newstr) - 1
	//jsonstr := newstr[:index] + "}" + newstr[index:]
	//listStrByte := []byte(str)
	//var megData []filter.MsgData
	//if err := json.Unmarshal(listStrByte, &megData); err != nil {
	//	log.Logger.Info("jsonObj format is error : {}", err)
	//}
	//if megData[0].InfrstructD.Msgtype == 200 {
	//	fmt.Println("productA find the character.")
	//	log.Logger.Info("productA find the character.")
	//	s.productA.Log("", str)
	//} else {
	//	log.Logger.Info("productB find the character.")
	//	s.productB.Log("", str)
	//}

	if find := strings.Contains(str, "\"msgtype\":200"); find {
		fmt.Println("productA find the character.")
		log.Logger.Info("productA find the character.")
		msg := &sarama.ProducerMessage{
			Topic: s.product.GetTopicA(),
			Key:   sarama.StringEncoder(""),
			Value: sarama.StringEncoder(str),
		}
		msga <- msg

	} else {
		fmt.Println("productB find the character.")
		msg := &sarama.ProducerMessage{
			Topic: s.product.GetTopicB(),
			Key:   sarama.StringEncoder(""),
			Value: sarama.StringEncoder(str),
		}
		msgb <- msg
	}

	<-s.limitChan
}

func (s *Server) UdpServer(address string, msga chan *sarama.ProducerMessage, msgb chan *sarama.ProducerMessage) {
	udpAddr, err := net.ResolveUDPAddr("udp", address)
	conn, err := net.ListenUDP("udp", udpAddr)
	log.Logger.Info("启动 udp 服务监听-----------")
	fmt.Println("Listen udp port : " + address)
	defer conn.Close()
	if err != nil {
		fmt.Println("read from connect failed, err:" + err.Error())
		log.Logger.Error("read from connect failed, err:{}", err)
		os.Exit(1)
	}

	for {
		s.limitChan <- true
		go s.udpProcess(conn, msga, msgb)
	}
}
